package sql;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.sql.Timestamp;

/**

 TopN 统计
 name:  用户
 val： 当前分数

 实时统计分数前3的用户
 思路：
 1. 数据流中name有重复的 结果中不应该有重复的
 2. 数据可能乱序


 eg. name有重复所以直接开窗会出现：
        a:10 b:20 c:30 b:40  -> b40,c30,b20 然而正确结果应该为 b40,c30,a10
 */
public class D11_1KafakSource_TopN {


    public static void main(String[] args) {


        Configuration flinkConf = new Configuration();
        flinkConf.setString("rest.port","9091");
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(flinkConf);
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);


        String genSql = "CREATE TABLE ods_tb ( " +
            " stime STRING," +
            " name STRING," +
            " val BIGINT," +
            " rowtime AS to_timestamp(stime)," +
            " WATERMARK FOR rowtime AS rowtime - interval '10' second" +
            ") WITH ( " +
            "  'connector' = 'kafka'," +
            "  'topic' = 'test'," +
            "  'properties.bootstrap.servers' = 'wsl:9092'," +
            "  'properties.group.id' = 'testGroup'," +
            "  'scan.startup.mode' = 'latest-offset'," +
            "  'format' = 'json'" +
            ")";
        //
        // String print = "CREATE TABLE print (" +"    " +
        //     "    name STRING, " +
        //     "    val BIGINT, " +
        //     "    rownum BIGINT " +
        //     ") WITH (" +
        //     "     'connector' = 'print'" +
        //     ")";

        /**
         * 放在redis就比较直观
         */
        String print = "CREATE TABLE print (" +
            "    snkey STRING, " +
            "    key STRING, " +
            "    val STRING, " +
            " primary key (snkey) not ENFORCED" +
            ") WITH (" +
            "    'connector' = 'redis',\n" +
            "    'host' = 'bdyun-suzhou-tw04f0163',\n" +
            "    'port' = '6379',\n" +
            "    'redis-mode' = 'single',\n" +
            "    'timeout' = '2000',\n" +
            "    'password' = 'yy2024',\n" +
            "    'command' = 'HMSET',\n" +
            "    'max.retries' = '5',\n" +
            "    'database' = '5',\n" +
            "    'ttl' = '10800'" +
            ")";

        /**
         * name分组 每个key取最新一条
         */
        // String sqlNew1 = "CREATE VIEW v1 " +
        //     " SELECT " +
        //     " name," +
        //     " val" +
        //     " FROM " +
        //     " (SELECT *,row_number() OVER(PARTITION BY name ORDER BY rowtime DESC) as rownum FROM ods_tb) " +
        //     " WHERE rownum <= 3";




        /**
         * 不分组 仅按val排序取Top N
         * 书中介绍了这种取max的办法，我觉得行不通
         * 比如输入数据：
         * +I[keyWord=a, hotNum=100]
         * +I[keyWord=b, hotNum=90]
         * +I[keyWord=c, hotNum=80]
         * +I[keyWord=a, hotNum=70]
         * 正确排序结果应该是：b,c,a
         * 但由于a只取最大的导致后续有更新的hotNum也无法去更新100
         */
        String sqlTopN = "insert into print " +
                " SELECT " +
            " 'topn' as snkey," +
            " CAST(rownum as STRING) key," +
            " concat(name,CAST(val as STRING)) as val" +
            " FROM " +
            " (SELECT *,row_number() OVER(ORDER BY val DESC) as rownum FROM " +
            "( SELECT name,max(val) as val FROM ods_tb GROUP BY name) " +
            ") " +
            " WHERE rownum <= 3";






        tableEnv.executeSql(genSql);
        tableEnv.executeSql(print);

        System.out.println(sqlTopN);
        tableEnv.executeSql(sqlTopN);


        System.out.println(new Timestamp(System.currentTimeMillis()));



    }
}
